package app.dwd;

import app.BaseSQLApp;
import common.Constant;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import utils.SQLUtil;

public class Dwd_03_DwdReviewInfo extends BaseSQLApp {
    public static void main(String[] args) {
        new Dwd_03_DwdReviewInfo().init(3002, 2, "Dwd_03_DwdReviewInfo");
    }

    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        readOdsDb(tEnv, "Dwd_03_DwdReviewInfo");
        Table reviewInfo = tEnv.sqlQuery("select  data['id'] id,  data['user_id'] user_id,  data['course_id'] course_id,  data['review_txt'] review_txt,  data['review_stars'] review_stars,  ts   from ods_db  where `database`='edu'  and `table`='review_info'  and `type`='insert' ");
        tEnv.createTemporaryView("review_info", reviewInfo);
        tEnv.executeSql("create table dwd_review_info(   id string,  user_id string,  course_id string,  review_txt string,  review_stars string,  ts bigint  )" + SQLUtil.getKafkaSink(Constant.TOPIC_DWD_DWD_REVIEW_INFO));
        reviewInfo.executeInsert("dwd_review_info");
    }
}